package com.enjoy.rocketmq.transaction;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class TransactionListenerImpl implements TransactionListener {

    private AtomicInteger integer = new AtomicInteger(0);


    private Map<String, Integer> localTrans = new ConcurrentHashMap<>();


    /**
     * 在提交事务消息后执行
     * 返回COMMIT_MESSAGE状态的消息会立即被消费者消费到
     * 返回ROLLBACK_MESSAGE状态的消息会被丢弃
     * 返回UNKNOWN的消息会由Broker过一段时间再来回查事务的状态
     *
     * @param msg msg
     * @param arg arg
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String tags = msg.getTags();
        if (tags.contains("tag_a")) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (tags.contains("tag_b")) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else {
            return LocalTransactionState.UNKNOW;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String tags = msg.getTags();
        if (tags.contains("tag_c")) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (tags.contains("tag_d")) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else {
            return LocalTransactionState.UNKNOW;
        }
    }
}
